package tinyiotserver

import (
	"bytes"
	"crypto/tls"
	"crypto/x509"
	"errors"
	"io"
	"io/ioutil"
	"net"
	"time"

	"gitee.com/mqyqingkong/tiny-iot-server/common"
)

type Client interface {
	// Close 关闭连接
	Close() error

	// Submit 发起请求，不需要等待服务器返回处理结果(并行安全)，
	// expireAfterNowMillis用于设置消息的过期时间，单位为ms。过期时间 = 服务端接收消息的时间 + expireAfterNowMillis，
	// 当expireAfterNowMillis=0时，提交的消息无过期时间。
	Submit(topic string, payload []byte, expireAfterNowMillis uint32) error

	// SubmitWithAsyncCallback 发起请求，不等待服务器返回处理结果，通过callback异步处理服务端响应(并行不安全)，
	// expireAfterNowMillis用于设置消息的过期时间，单位为ms。过期时间 = 服务端接收消息的时间 + expireAfterNowMillis，
	// 当expireAfterNowMillis=0时，提交的消息无过期时间。
	// callback 异步处理服务端响应。
	SubmitWithAsyncCallback(topic string, payload []byte, expireAfterNowMillis uint32, callback func(msg *common.Msg, occuredErr error)) error

	// SubmitWaitReply 发起请求，需要等待服务器返回处理结果(并行不安全)
	// expireAfterNowMillis用于设置消息的过期时间，单位为ms。过期时间 = 服务端接收消息的时间 + expireAfterNowMillis
	// 当expireAfterNowMillis=0时，提交的消息无过期时间
	SubmitWaitReply(topic string, payload []byte, expireAfterNowMillis uint32) (*common.Msg, error)

	// Query 查询topic对应的消息，不删除消息(并行不安全)
	Query(topic string, payload []byte) (*common.Msg, error)

	// QueryPop查询topic对应的消息，并删除消息(并行不安全)
	QueryPop(topic string, payload []byte) (*common.Msg, error)

	// QueryStream 发起流请求，通过Receive接收服务器返回处理结果，一个消息只能被一个client查询(并行不安全)
	QueryStream(topic string, payload []byte) (receive func() (*common.Msg, error), err error)

	// QueryStreamGroup 发起流请求，通过Receive接收服务器返回处理结果，每个消息可以被多个client查询(并行不安全)
	QueryStreamGroup(topic string, payload []byte) (receive func() (*common.Msg, error), err error)

	// Auth 请求授权
	Auth(payload []byte) error

	SetTimeOut(timeout time.Duration)
}

type TimeoutReadWriteCloser interface {
	io.ReadWriteCloser
	SetReadTimeout(timeout time.Duration)
	SetWriteTimeout(timeout time.Duration)
}

type netReadWriteCloser struct {
	conn net.Conn
}

func (rwc *netReadWriteCloser) Read(p []byte) (n int, err error) {
	return rwc.conn.Read(p)
}

func (rwc *netReadWriteCloser) Write(p []byte) (n int, err error) {
	return rwc.conn.Write(p)
}

func (rwc *netReadWriteCloser) Close() error {
	return rwc.conn.Close()
}

func (rwc *netReadWriteCloser) SetReadTimeout(timeout time.Duration) {
	if timeout == 0 {
		rwc.conn.SetReadDeadline(time.Time{})
	} else {
		rwc.conn.SetReadDeadline(time.Now().Add(timeout))
	}

}
func (rwc *netReadWriteCloser) SetWriteTimeout(timeout time.Duration) {
	if timeout == 0 {
		rwc.conn.SetWriteDeadline(time.Time{})
	} else {
		rwc.conn.SetWriteDeadline(time.Now().Add(timeout))
	}
}

type client struct {
	clientId uint32
	conn     TimeoutReadWriteCloser
	timeout  time.Duration
	authCode string
}

func NewClient(serverAddress string, clientId uint32) (Client, error) {
	conn, err := net.Dial("tcp", serverAddress)
	if err != nil {
		return nil, err
	}
	c := NewClientByConn(&netReadWriteCloser{
		conn: conn,
	}, clientId)
	return c, nil
}

func NewUnixClient(serverAddress string, clientId uint32) (Client, error) {
	conn, err := net.Dial("unix", serverAddress)
	if err != nil {
		return nil, err
	}
	c := NewClientByConn(&netReadWriteCloser{
		conn: conn,
	}, clientId)
	return c, nil
}

type TLSConfig struct {
	RootCACertPath string
	CertFilePath   string
	KeyFilePath    string
}

func NewClientWithTLS(serverAddress string, clientId uint32, tlsConf *TLSConfig) (Client, error) {
	if tlsConf == nil {
		return nil, errors.New("tls configuration must be specified")
	}
	pool := x509.NewCertPool()
	caCrt, err := ioutil.ReadFile(tlsConf.RootCACertPath)
	if err != nil {
		return nil, err
	}
	pool.AppendCertsFromPEM(caCrt)

	cliCrt, err := tls.LoadX509KeyPair(tlsConf.CertFilePath, tlsConf.KeyFilePath)
	if err != nil {
		return nil, err
	}

	conf := &tls.Config{
		RootCAs:      pool,
		Certificates: []tls.Certificate{cliCrt},
	}
	conn, err := tls.Dial("tcp", serverAddress, conf)
	if err != nil {
		return nil, err
	}
	c := NewClientByConn(&netReadWriteCloser{
		conn: conn,
	}, clientId)
	return c, nil
}

func NewClientByConn(conn TimeoutReadWriteCloser, clientId uint32) Client {
	c := &client{
		clientId: clientId,
		conn:     conn,
		timeout:  10 * time.Second,
		authCode: "",
	}
	return c
}

func (c *client) SetTimeOut(timeout time.Duration) {
	c.timeout = timeout
}
func (c *client) Submit(topic string, payload []byte, expireAfterNowMillis uint32) error {
	msg, err := c.createMessage(common.POST_NO_RESPONSE, topic, payload, expireAfterNowMillis)
	if err != nil {
		return err
	}
	mb := msg.Bytes()
	c.conn.SetWriteTimeout(c.timeout)
	_, err = c.conn.Write(mb)
	return err
}
func (c *client) SubmitWithAsyncCallback(topic string, payload []byte, expireAfterNowMillis uint32, callback func(msg *common.Msg, occuredErr error)) error {
	msg, err := c.createMessage(common.POST_WAIT_RESPONSE, topic, payload, expireAfterNowMillis)
	if err != nil {
		return err
	}
	mb := msg.Bytes()
	c.conn.SetWriteTimeout(c.timeout)
	_, err = c.conn.Write(mb)
	if err != nil {
		return err
	}
	go func() {
		c.conn.SetReadTimeout(c.timeout)
		replyMsg, err := common.ReadMsg(c.conn)
		callback(replyMsg, err)
	}()
	return nil
}

func (c *client) QueryStream(topic string, payload []byte) (receive func() (*common.Msg, error), err error) {
	msg, err := c.createMessage(common.QUERY_STREAM, topic, payload, 0)
	if err != nil {
		return nil, err
	}
	mb := msg.Bytes()
	c.conn.SetWriteTimeout(c.timeout)
	_, err = c.conn.Write(mb)
	if err != nil {
		return nil, err
	}
	receive = func() (*common.Msg, error) {
		c.conn.SetReadTimeout(0)
		replyMsg, err := common.ReadMsg(c.conn)
		return replyMsg, err
	}
	return receive, nil
}

func (c *client) QueryStreamGroup(topic string, payload []byte) (receive func() (*common.Msg, error), err error) {
	msg, err := c.createMessage(common.QUERY_STREAM_GROUP, topic, payload, 0)
	if err != nil {
		return nil, err
	}
	mb := msg.Bytes()
	c.conn.SetWriteTimeout(c.timeout)
	_, err = c.conn.Write(mb)
	if err != nil {
		return nil, err
	}
	receive = func() (*common.Msg, error) {
		c.conn.SetReadTimeout(0)
		replyMsg, err := common.ReadMsg(c.conn)
		return replyMsg, err
	}
	return receive, nil
}

func (c client) SubmitWaitReply(topic string, payload []byte, expireAfterNowMillis uint32) (*common.Msg, error) {
	msg, err := c.createMessage(common.POST_WAIT_RESPONSE, topic, payload, expireAfterNowMillis)
	if err != nil {
		return nil, err
	}
	return c.doSubmitWaitReply(msg)
}

func (c *client) doSubmitWaitReply(msg *common.Msg) (*common.Msg, error) {
	mb := msg.Bytes()
	c.conn.SetWriteTimeout(c.timeout)
	_, err := c.conn.Write(mb)
	if err != nil {
		return nil, err
	}
	c.conn.SetReadTimeout(c.timeout)
	replyMsg, err := common.ReadMsg(c.conn)
	return replyMsg, err
}

func (c client) Query(topic string, payload []byte) (*common.Msg, error) {
	msg, err := c.createMessage(common.QUERY, topic, payload, 0)
	if err != nil {
		return nil, err
	}
	return c.doSubmitWaitReply(msg)
}

func (c client) QueryPop(topic string, payload []byte) (*common.Msg, error) {
	msg, err := c.createMessage(common.QUERY_POP, topic, payload, 0)
	if err != nil {
		return nil, err
	}
	return c.doSubmitWaitReply(msg)
}

func (c client) createMessage(msgType common.MsgType, topic string, payload []byte, expireAfterNowMillis uint32) (*common.Msg, error) {
	msg := common.NewMsg(common.MsgHeader{
		Version:        common.Version,
		Type:           msgType,
		ClientId:       c.clientId,
		ExpireAfterNow: expireAfterNowMillis,
	}, common.MsgBody{
		Topic:   []byte(topic),
		Payload: payload,
	})
	if len(c.authCode) > 0 {
		msg.Body.Payload = append([]byte(c.authCode), payload...)
		msg.InitLength()
	}
	return msg, nil
}

func (c client) Close() error {
	return c.conn.Close()
}

func (c *client) Auth(payload []byte) error {
	resAuth, err := c.SubmitWaitReply("REQ_AUTH", payload, 0)
	if err != nil {
		return err
	}
	if isReqFail(resAuth) {
		return errors.New("Auth Fail: " + string(resAuth.Body.Payload))
	}
	c.clientId = resAuth.Header.ClientId
	c.authCode = string(resAuth.Body.Payload)
	return nil
}

func isReqFail(resp *common.Msg) bool {
	return bytes.HasPrefix(resp.Body.Payload, []byte("FAIL:"))
}
